package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.data_etl.udfs.TimeTruncUdf;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job2_流量olap分析主题轻度聚合模型 {
    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射dwd层的kafka中的行为明细数据表
        //  创建kafka中的维度打宽的行为日志的明细数据，逻辑映射表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                          "
                        +"     user_id           BIGINT,                     "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_Id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "
                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "
                        +"     url_prefix    STRING,                         "
                        +"     page_type    STRING,                          "
                        +"     page_service STRING,                          "

                        +"     proc_time AS proctime(),                      " // processing time 时间语义
                        +"     rt AS  to_timestamp_ltz(event_time,3),        " // 表达式字段，用于将event_time转成timestamp(3)类型
                        +"     WATERMARK FOR rt AS  rt - INTERVAL '0' SECOND   " // 基于rt字段定义watermark，从此，rt字段就具备了 flink中事件时间语义

                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd_events',                          "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'latest-offset',           "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           "
        );


        // 建表，映射目标表（doris中dws层的目标表）
        tenv.executeSql(
                " create table  if not exists tfc_olap_agg_doris (  " +
                        "     dt              DATE,                         " +
                        "     time_60m        STRING,                       " +
                        "     time_30m        STRING,                       " +
                        "     time_10m        STRING,                       " +
                        "     time_5m         STRING,                       " +
                        "     time_1m         STRING,                       " +
                        "     gps_province    STRING,                       " +
                        "     gps_city        STRING,                       " +
                        "     gps_region      STRING,                       " +
                        "     release_channel STRING,                       " +
                        "     device_type     STRING,                       " +
                        "     page_type       STRING,                       " +
                        "     page_service    STRING,                       " +
                        "     is_new          INT,                          " +
                        "     pv_cnt          bigint,                       " +
                        "     uv_cnt          bigint,                       " +  // 此处其实只是： user_id
                        "     session_cnt     STRING                        " +  // 此处其实只是： session_id
                        " ) WITH (                                          " +
                        "    'connector' = 'doris',                         " +
                        "    'fenodes' = 'doitedu:8030',                    " +
                        "    'table.identifier' = 'dws.tfc_overview_1m',    " +
                        "    'username' = 'root',                           " +
                        "    'password' = 'root',                           " +
                        "    'sink.label-prefix' = 'doris_label-"+System.currentTimeMillis()+"', " +
                        "    'sink.properties.columns' = 'dt,time_60m,time_30m,time_10m,time_5m,time_1m,gps_province,gps_city,gps_region,release_channel,device_type,page_type,page_service,is_new,pv_cnt,uv_cnt,session_cnt,uv_cnt=to_bitmap(uv_cnt),session_cnt=hll_hash(session_cnt) '  " +
                        " )                                                 "
        );


        tenv.createTemporaryFunction("time_trunc", TimeTruncUdf.class);
        tenv.executeSql(
                "INSERT INTO tfc_olap_agg_doris                                                       " +
                        "WITH tmp AS (                                                                          "+
                        "     SELECT                                                                            "+
                        "          user_id                                                                      "+
                        "     	  ,session_id                                                                   "+
                        "     	  ,event_time                                                                   "+
                        "         ,gps_province                                                                 "+
                        "         ,gps_city                                                                     "+
                        "         ,gps_region                                                                   "+
                        "         ,release_channel                                                              "+
                        "         ,device_type                                                                  "+
                        "         ,page_type                                                                    "+
                        "         ,page_service                                                                 "+
                        "         ,if(date_format(register_time,'yyyy-MM-dd') = CURRENT_DATE,1,0) as  is_new    "+
                        "         ,rt  		                                                                    "+
                        "     FROM  dwd_kafka                                                                   "+
                        "     WHERE event_Id = 'page_load'                                                      "+
                        " )                                                                                     "+
                        "                                                                                       "+
                        " SELECT                                                                                "+
                        " 	TO_DATE(date_format(window_end,'yyyy-MM-dd'))   as  dt                              "+
                        " 	,time_trunc(window_end,60)  as time_60m                                             "+
                        " 	,time_trunc(window_end,30)  as time_30m                                             "+
                        " 	,time_trunc(window_end,10)  as time_10m                                             "+
                        " 	,time_trunc(window_end,5)   as time_5m                                              "+
                        " 	,time_trunc(window_end,1)   as time_1m                                              "+
                        "   ,gps_province                                                                       "+
                        "   ,gps_city                                                                           "+
                        "   ,gps_region                                                                         "+
                        "   ,release_channel                                                                    "+
                        "   ,device_type                                                                        "+
                        "   ,page_type                                                                          "+
                        "   ,page_service                                                                       "+
                        " 	,is_new                                                                             "+
                        " 	,count(1) as pv_cnt                                                                 "+
                        " 	,user_id  as uv_cnt                                                                 "+
                        " 	,session_id as session_cnt                                                          "+
                        "                                                                                       "+
                        " FROM  TABLE(                                                                          "+
                        "     TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)                              "+
                        " )                                                                                     "+
                        " GROUP BY                                                                              "+
                        "     window_start                                                                      "+
                        " 	  ,window_end                                                                       "+
                        "     ,user_id	                                                                        "+
                        "     ,session_id                                                                       "+
                        "     ,gps_province                                                                     "+
                        "     ,gps_city                                                                         "+
                        "     ,gps_region                                                                       "+
                        "     ,release_channel                                                                  "+
                        "     ,device_type                                                                      "+
                        "     ,page_type                                                                        "+
                        "     ,page_service                                                                     "+
                        "     ,is_new                                                                            "

        );

    }
}
